package com.sleep.satokendemo.listener;

import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.listener.MessageListener;
import org.redisson.api.stream.StreamAddArgs;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author GuoAo
 * @className RedisListener
 * @date 2025/6/30
 * @description TODO
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class RedisListener{
    private final RedissonClient redissonClient;
    private final RstreamTopicListener rstreamTopicListener;
    @PostConstruct
    public void init(){
//        RStream<Object, Object> stream = redissonClient.getStream("log-stream");
//        // stream.createGroup("myGroup", StreamMessageId.NEWEST);
//        while(true){
//            Map<StreamMessageId, Map<Object, Object>> streamMessageIdMapMap = stream.readGroup("myGroup", "myConsumer",10,10, TimeUnit.SECONDS);
//            log.info("streamMessageIdMapMap============================>{}",streamMessageIdMapMap);
//            for(Map.Entry<StreamMessageId, Map<Object, Object>> entry : streamMessageIdMapMap.entrySet()){
//                log.info("messageId:{}, message:{}",entry.getKey(),entry.getValue());
//                stream.ack("myGroup", entry.getKey());
//            }
//        }
        RTopic topic = redissonClient.getTopic("log-Topic");
        topic.addListener(String.class, rstreamTopicListener);
    }

    public void push(String message){
        RTopic topic = redissonClient.getTopic("log-Topic");
        topic.publish(message);
        RStream<Object, Object> stream = redissonClient.getStream("log-stream");
        stream.add(StreamMessageId.AUTO_GENERATED, StreamAddArgs.entry("message", message));
    }
}

